package com.zhss.microservice.server.network;

import com.alibaba.fastjson.JSONObject;
import com.zhss.microservice.server.constant.NodeStatus;
import com.zhss.microservice.server.node.ControllerVote;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.zhss.microservice.server.constant.MessageType;

import java.nio.ByteBuffer;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * master节点的消息接收组件（线程）
 *
 * 1、不停的从网络通信组件的receiver队列里获取最新的消息
 * 2、判断消息的类型，对消息进行对象转化
 * 3、把消息推送到不同的业务模块所对应的队列里去
 * 4、对各种不同的业务模块提供获取自己业务的消息的接口
 *
 */
public class ServerMessageReceiver extends Thread {

    static final Logger LOGGER = LoggerFactory.getLogger(ServerMessageReceiver.class);

    /**
     * master节点的网络通信组件
     */
    private ServerNetworkManager masterNetworkManager;
    /**
     * 投票消息接收队列
     */
    private LinkedBlockingQueue<ControllerVote> voteReceiveQueue =
            new LinkedBlockingQueue<ControllerVote>();
    /**
     * 槽位数据接收队列
     */
    private LinkedBlockingQueue<ConcurrentHashMap<Integer, String>> slotsAllocationReceiveQueue =
            new LinkedBlockingQueue<>();
    /**
     * 自己负责的槽位范围的消息接收队列
     */
    private LinkedBlockingQueue<String> nodeSlotsQueue =
            new LinkedBlockingQueue<>();

    /**
     * 初始化方法
     * @param masterNetworkManager
     */
    public ServerMessageReceiver(ServerNetworkManager masterNetworkManager) {
        this.masterNetworkManager = masterNetworkManager;
    }

    /**
     * 线程主循环体
     */
    public void run() {
        while(NodeStatus.isRunning()) {
            try {
                ByteBuffer message = masterNetworkManager.takeMessage();
                int messageType = message.getInt();

                // 如果是controller投票消息
                if (messageType == MessageType.VOTE) {
                    ControllerVote vote = new ControllerVote(message);
                    voteReceiveQueue.put(vote);
                    LOGGER.info("receive a vote message: " + vote);
                }
                // 如果是槽位分配消息
                else if (messageType == MessageType.SLOTS_ALLOCATION) {
                    int remaining = message.remaining();
                    byte[] bytes = new byte[remaining];
                    message.get(bytes);

                    String slotsAllocationJSON = new String(bytes);
                    ConcurrentHashMap<Integer, String> slotsAllocation = JSONObject.parseObject(
                            slotsAllocationJSON, ConcurrentHashMap.class);
                    slotsAllocationReceiveQueue.put(slotsAllocation);

                    LOGGER.info("receive a slots allocation message: " + slotsAllocation);
                }
                // 如果是负责槽位范围消息
                else if(messageType == MessageType.NODE_SLOTS) {
                    int remaining = message.remaining();
                    byte[] bytes = new byte[remaining];
                    message.get(bytes);

                    String slots = new String(bytes);
                    nodeSlotsQueue.put(slots);

                    LOGGER.info("receive a node slots message: " + slots);
                }
            } catch(Exception e) {
                LOGGER.error("receive message error......", e);
            }
        }
    }

    /**
     * 阻塞式获取投票消息
     * @return
     */
    public ControllerVote takeVote() {
        try {
            return voteReceiveQueue.take();
        } catch(Exception e) {
            LOGGER.error("take vote message error......", e);
            return null;
        }
    }

    /**
     * 阻塞式获取槽位分配消息
     * @return
     */
    public ConcurrentHashMap<Integer, String> takeSlotsAllocation() {
        try {
            return slotsAllocationReceiveQueue.take();
        } catch(Exception e) {
            LOGGER.error("take slots allocation message error......", e);
            return null;
        }
    }

    /**
     * 阻塞式获取负责的槽位范围
     * @return
     */
    public String takeNodeSlots() {
        try {
            return nodeSlotsQueue.take();
        } catch(Exception e) {
            LOGGER.error("take node slots message error......", e);
            return null;
        }
    }

}
